package com.example.message_queue.mqserver;

import com.example.message_queue.common.Consumer;
import com.example.message_queue.common.MqException;
import com.example.message_queue.mqserver.core.*;
import com.example.message_queue.mqserver.datacenter.DiskDataCenter;
import com.example.message_queue.mqserver.datacenter.MemoryDataCenter;
import lombok.Getter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with Intellij IDEA.
 * Description;
 * User:TQ02
 * Data:2024-04-03
 * Time:16:09
 * Text：虚拟主机
 * 管理交换机、队列、绑定等
 */
public class VirtualHost {
    private String virtualHostHome;
    @Getter
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    @Getter
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router =new Router();
    private ConsumerManager consumerManager =new ConsumerManager(this);

    // 操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    // 操作队列的锁对象
    private final Object queueLocker = new Object();

    public VirtualHost(String name){
        this.virtualHostHome = name;

        diskDataCenter.init();
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost]:内存恢复失败");
        }
    }

    //创建交换机
    public boolean exchangeDeclare(String exchangeName, ExchangeType type,boolean durable,boolean autoDelete,
                                   Map<String,Object> argument){
        exchangeName = virtualHostHome + exchangeName;
        try{
            synchronized (exchangeLocker) {
                //判断交换机是否存在
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    System.out.println("交换机已经存在");
                    return true;
                }
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(type);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(argument);
                //写入硬盘
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                //写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost]:交换机创建成功，ExchangeName:" + exchangeName);
                //先写硬盘和内存是有讲究的，硬盘更容易失败
            }
            return true;
        }catch(Exception e){
            System.out.println("[VirtualHost]:交换机创建失败，ExchangeName:"+exchangeName);
            e.printStackTrace();
            return false;
        }
    }
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostHome + exchangeName;
        try{
            synchronized (exchangeLocker){
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if(toDelete == null){
                    throw  new MqException("[VirtualHost]:交换机不存在，ExchangeName:"+exchangeName);
                }
                if(toDelete.getDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost]:交换机删除成功，ExchangeName:"+exchangeName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost]:交换机删除失败，ExchangeName:"+exchangeName);
            e.printStackTrace();
            return false;
        }
    }
    public boolean createQueue(String queueName,boolean durable,boolean exclusive,boolean autoDelete,
                               Map<String,Object> arguments){
        queueName = virtualHostHome +queueName;
        try{
            synchronized (queueLocker){
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue != null){
                    System.out.println("[VirtualHost]:队列存在，ExchangeName:"+queueName);
                    return false;
                }
                MSGQueue msgQueue =new MSGQueue();
                msgQueue.setAutoDelete(autoDelete);
                msgQueue.setExclusive(exclusive);
                msgQueue.setName(queueName);
                msgQueue.setArguments(arguments);
                //插入硬盘
                diskDataCenter.insertQueue(msgQueue);
                memoryDataCenter.insertQueue(msgQueue);
                System.out.println("[VirtualHost]:对列创建成功，QueueName:"+queueName);
                //先写硬盘和内存是有讲究的，硬盘更容易失败
            }
            return true;
        } catch (IOException e) {
            System.out.println("[VirtualHost]:队列创建失败，QueueName:"+queueName);
            e.printStackTrace();
            return false;
        }
    }
    public boolean deleteQueue(String queueName){
        queueName = virtualHostHome + queueName;
        try{
            synchronized (queueLocker) {
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue == null) {
                    throw new MqException("[VirtualHost]:对列已存在，QueueName:" + queueName);
                }
                if (existsQueue.isDurable()) {
                    diskDataCenter.deleteQueue(existsQueue);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost]:对列删除成功，QueueName:" + queueName);
            }
            return true;
        } catch (Exception e){
            System.out.println("[VirtualHost]:对列删除失败，QueueName:"+queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean createBinding(String exchangeName,String queueName,String bindingKey){
        queueName = virtualHostHome + queueName;
        exchangeName = virtualHostHome + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 判定当前的绑定是否已经存在了.
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        throw new MqException("[VirtualHost] binding 已经存在! queueName=" + queueName
                                + ", exchangeName=" + exchangeName);
                    }
                    // 2. 验证 bindingKey 是否合法.
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey 非法! bindingKey=" + bindingKey);
                    }
                    // 3. 创建 Binding 对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    // 4. 获取一下对应的交换机和队列. 如果交换机或者队列不存在, 这样的绑定也是无法创建的.
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在! queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
                    }
                    // 5. 先写硬盘
                    if (queue.isDurable() && exchange.getDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    // 6. 写入内存
                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    public boolean queueUnbind(String exchangeName, String queueName) {
        queueName = virtualHostHome + queueName;
        exchangeName = virtualHostHome+ exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 获取 binding 看是否已经存在~
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("[VirtualHost] 删除绑定失败! 绑定不存在! exchangeName=" + exchangeName + ", queueName=" + queueName);
                    }
                    // 2. 无论绑定是否持久化了, 都尝试从硬盘删一下. 就算不存在, 这个删除也无副作用.
                    diskDataCenter.deleteBinding(binding);
                    // 3. 删除内存的数据
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功!");
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败!");
            e.printStackTrace();
            return false;
        }
    }
        //发送消息到指定队列和交换机里
    public boolean basicPublish(String exchangeName,String routingKey,BasicProertirs basicProertirs,
                                byte[] body){
        //转换交换机名字
        exchangeName = virtualHostHome + exchangeName;
        try{
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] rountKey非法");
            }
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null){
                throw new MqException("[VirtualHost] 交换机不存在");
            }
            if(exchange.getType() == ExchangeType.DIRECT){
                //在DIRECT类型里，routingKey就是队列名字，获取队列名字
                String queueName = virtualHostHome + routingKey;
                //构造消息
                Message message = Message.createMessageWithId(routingKey,basicProertirs,body);
                //查找消息对应的队列
                MSGQueue queue =memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost] 消息队列不存在");
                }
                //队列存在时
                sendMessage(queue,message);
            }else{
                //找到与该交换机的所有绑定
                ConcurrentHashMap<String,Binding> bindingMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String,Binding> entry : bindingMap.entrySet()){
                    //根据获取到的绑定对象，判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        // 此处咱们就不抛出异常了. 可能此处有多个这样的队列.
                        // 希望不要因为一个队列的失败, 影响到其他队列的消息的传输.
                        System.out.println("[VirtualHost] basicPublish 发送消息时, 发现队列不存在! queueName=" + binding.getQueueName());
                        continue;
                    }
                    // 2) 构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProertirs, body);
                    // 3) 判定这个消息是否能转发给该队列.
                    //    如果是 fanout, 所有绑定的队列都要转发的.
                    //    如果是 topic, 还需要判定下, bindingKey 和 routingKey 是不是匹配.
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    // 4) 真正转发消息给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 消息发送失败!");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws InterruptedException {

        int deliverMode = message.getDeliverMode();
        if(deliverMode == 1){
            try {
                diskDataCenter.sendMessage(queue,message);
            } catch (IOException | MqException e) {
                e.printStackTrace();
            }
        }
        memoryDataCenter.sendMessage(queue,message);
        //通知消费者可以消费消息
        consumerManager.notifyConsume(queue.getName());
    }

    //通知消费者消费消息！
    /*
    consumerTag:消费者的身份标识
    autoAck:消息被消费者完成之后，应答的方式，true为自动应答，false为手动应答
    Consumer:是一个回调函数，函数式接口
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        //构造一个ConsumerEnv对象，把这个队列找到，再把这个Consumer对象添加到该队列
        queueName = virtualHostHome + queueName;
        try{
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsume成功");
            return true;
        }catch(Exception e){
            System.out.println("[VirtualHost] basicConsume失败");
            return false;
        }
    }

    public boolean basicAck(String queueName,String messageId){
        queueName = virtualHostHome + queueName;
        try{
            Message message = memoryDataCenter.selectMessage(messageId);
            if(message == null){
                throw new MqException("[VirtualHost] basicAck:消息获取失败"+ messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                throw new MqException("[VirtualHost] basicAck:消息队列获取失败"+queueName);
            }
            //删除硬盘数据
            if(message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue,message);
            }
            //删除消息中心的数据
            memoryDataCenter.deleteMessage(messageId);
            //删除待确认的数据
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            System.out.println("[VirtualHost] basicAck:消息成功确认");
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck:消息确认失败");
           return false;
        }
    }
}
